
#if !defined __RTCN_RUDPFLOWCONTROL_H__
#define __RTCN_RUDPFLOWCONTROL_H__

#include "RtBase.h"
#include "RtUtilTemplates.h"
#include "RtHashMapT.h"
#include "RtPairInetAddr.h"
#include "RtReactorInterface.h"
#include "RtCnRudp.h"

#define RUDP_CONNS_USE_ARRAY

#define RUDP_CONNS_ALLOCSIZE_INIT 1024

class CRtRudpFlowControlReleaseEvent : public IRtEvent
{
public:
	CRtRudpFlowControlReleaseEvent(CRtRudpFlowControl* inInstance)
		: m_instance(inInstance)
	{
	}
	virtual ~CRtRudpFlowControlReleaseEvent()
	{
	}

	virtual RtResult OnEventFire();

private:
	CRtRudpFlowControl* m_instance;
};


class CRtRudpFlowControl: public IRtTimerHandler
{
#ifdef RUDP_CONNS_USE_ARRAY
	typedef CRtHashMapT<CRtPairInetAddr, unsigned > RudpConnIndexesType;
#else
	typedef CRtHashMapT<CRtPairInetAddr, CRtAutoPtr<CRtRudpConn> > RudpConnsType;
#endif

public:
	static CRtRudpFlowControl* Instance()
	{
		return CRtSingletonT<CRtRudpFlowControl>::Instance();
	}
	
	RtResult InsertConn(CRtRudpConn* pConn) 
	{
// 		RT_ASSERTE(CRtThreadManager::IsEqualCurrentThread(m_pThreadNetwork->GetThreadId()));
#ifdef RUDP_CONNS_USE_ARRAY
		CRtPairInetAddr addrPair(pConn->GetPeerAddr(), pConn->GetLocalAddr());
		RudpConnIndexesType::value_type nodeNew(addrPair, m_connArraySize);
		pair<RudpConnIndexesType::iterator, bool> rv = m_connIndexes.insert(nodeNew);
		if (rv.second==false)
		{
			unsigned oldID = rv.first->second;
			CRtRudpConn* pOldConn = m_connArray[oldID];
			if (pOldConn->m_status!=RUDP_STATE_CLOSED)
			{
				RUDP_WARNING_TRACE("CRtRudpFlowControl::InsertConn, duplicated connection! conn1="<<pOldConn<<" state="<<pOldConn->m_status<<" conn2="<<pConn);
				pOldConn->OnDisconnect(RT_ERROR_NETWORK_CONNECTION_LOST, pOldConn->GetLowerTransport());
			}
			pOldConn->ReleaseReference();
			m_connArray[oldID] = pConn;
		}
		else
		{
			if (m_connArraySize==m_connArrayAllocSize)
			{
				if ( m_connArrayAllocSize == 0 )
					m_connArrayAllocSize = RUDP_CONNS_ALLOCSIZE_INIT;
				else
					m_connArrayAllocSize *= 2;

/*
				CRtRudpConn** new_array = new CRtRudpConn* [ m_connArrayAllocSize ];
				memcpy(new_array, m_connArray, m_connArraySize*sizeof(CRtRudpConn*));
				delete[] m_connArray;
				m_connArray = new_array;
*/
				m_connArray = (CRtRudpConn**)realloc(m_connArray, m_connArrayAllocSize*sizeof(CRtRudpConn*));
				RT_ASSERTE(m_connArray);
			}

			// Insert the new item at the correct spot
			m_connArray[ m_connArraySize++ ] = pConn;
		}

		pConn->AddReference();
#else
		CRtPairInetAddr addrPair(pConn->GetPeerAddr(), pConn->GetLocalAddr());
		RudpConnsType::value_type nodeNew(addrPair, pConn);
		pair<RudpConnsType::iterator, bool> rv = m_conns.insert(nodeNew);
		if (rv.second==false)
		{
			CRtRudpConn* pOldConn = rv.first->second.Get();
			if (pOldConn->GetCurrStatus()!=RUDP_STATE_CLOSED)
			{
				RUDP_WARNING_TRACE("CRtRudpFlowControl::InsertConn, duplicated connection! conn1="<<pOldConn<<" state="<<pOldConn->GetCurrStatus()<<" conn2="<<pConn);
				pOldConn->OnDisconnect(RT_ERROR_NETWORK_CONNECTION_LOST, pOldConn->GetTPTransport());
			}
			m_conns.erase(rv.first);
			rv = m_conns.insert(nodeNew);
			RT_ASSERTE(rv.second==true);
		}
#endif

		if (!m_bScheduled)
		{
			m_pReactor->ScheduleTimer(this, NULL, CRtTimeValue(0, g_dwDefaultTimerTickInterval*1000), 0);
			m_bScheduled = TRUE;
		}

		return RT_OK;
	}

	void RemoveConn(CRtRudpConn* pConn)
	{
//		RT_ASSERTE(CRtThreadManager::IsEqualCurrentThread(m_pThreadNetwork->GetThreadId()));
		CRtPairInetAddr rmAddrPair(pConn->GetPeerAddr(), pConn->GetLocalAddr());
#ifdef RUDP_CONNS_USE_ARRAY
		RudpConnIndexesType::iterator it = m_connIndexes.find(rmAddrPair);
		if (it!=m_connIndexes.end())
		{
			unsigned connID = it->second;
			m_connIndexes.erase(it);
			m_connArray[connID]->ReleaseReference();
			if (connID<m_connArraySize-1)
			{
				m_connArray[connID] = m_connArray[m_connArraySize-1];
				CRtPairInetAddr addrPair(m_connArray[connID]->GetPeerAddr(), m_connArray[connID]->GetLocalAddr());
				m_connIndexes[addrPair] = connID;
			}
			--m_connArraySize;

			if (m_connArraySize==0 && m_bScheduled)
			{
				m_pReactor->CancelTimer(this);
				m_bScheduled = FALSE;
			}
		}
#else
		RudpConnsType::iterator it = m_conns.find(rmAddrPair);
		if (it!=m_conns.end())
		{
			RemoveConn(it);
		}
#endif
	}

protected:
	friend class CRtSingletonT<CRtRudpFlowControl>;
	friend class CRtRudpFlowControlReleaseEvent;

	CRtRudpFlowControl()
#ifdef RUDP_CONNS_USE_ARRAY
	: m_connIndexes(RUDP_CONNS_ALLOCSIZE_INIT)
	, m_connArraySize(0)
	, m_nextConnIndex(0)
#else
	: m_conns(RUDP_CONNS_ALLOCSIZE_INIT)
#endif
	, m_bScheduled(FALSE)
	, m_internalPacketPool(new CRtRudpPacketPool)
	{
		RUDP_INFO_TRACE_THIS("CRtRudpFlowControl::CRtRudpFlowControl");
		m_pThreadNetwork = CRtThreadManager::Instance()->GetThread(CRtThreadManager::TT_NETWORK);
		m_pReactor = CRtThreadManager::Instance()->GetThreadReactor(CRtThreadManager::TT_NETWORK);
		RT_ASSERTE(m_pThreadNetwork);
		RT_ASSERTE(m_pReactor);
		RT_ASSERTE(CRtThreadManager::IsEqualCurrentThread(m_pThreadNetwork->GetThreadId()));

#ifdef RUDP_CONNS_USE_ARRAY
		m_connArrayAllocSize = RUDP_CONNS_ALLOCSIZE_INIT;
//		m_connArray = new CRtRudpConn* [m_connArrayAllocSize];
		m_connArray = (CRtRudpConn**)malloc(m_connArrayAllocSize*sizeof(CRtRudpConn*));
		RT_ASSERTE(m_connArray);
#endif
	}
	virtual ~CRtRudpFlowControl()
	{
		RUDP_INFO_TRACE_THIS("CRtRudpFlowControl::~CRtRudpFlowControl");
		CRtRudpFlowControlReleaseEvent* event = new CRtRudpFlowControlReleaseEvent(this);
		m_pReactor->SendEvent(event);
	}

	/// Release() should be call in network thread.
	void Release()
	{
#ifdef RUDP_CONNS_USE_ARRAY
		RUDP_INFO_TRACE_THIS("CRtRudpFlowControl::Release, m_connArraySize="<<m_connArraySize);
		for (unsigned i=0; i<m_connArraySize; ++i)
		{
			/// Call rudp::Release() to avoid crash.
			m_connArray[i]->Release();
			m_connArray[i]->ReleaseReference();
		}
//		delete[] m_connArray;
		free(m_connArray);
		m_connArrayAllocSize = 0;
		m_connArraySize = 0;
#endif
		delete m_internalPacketPool;

		if (m_bScheduled)
		{
			m_pReactor->CancelTimer(this);
			m_bScheduled = FALSE;
		}
	}

#ifdef RUDP_CONNS_USE_ARRAY
/*
	void RemoveConn(unsigned i)
	{
		CRtPairInetAddr rmAddrPair(m_connArray[i]->GetPeerAddr(), m_connArray[i]->GetLocalAddr());
		m_connIndexes.erase(rmAddrPair);
		m_connArray[i]->ReleaseReference();
		if (i<m_connArraySize-1)
		{
			m_connArray[i] = m_connArray[m_connArraySize-1];
		}
		--m_connArraySize;
	}
*/
#else
	void RemoveConn(RudpConnsType::iterator& it)
	{
		m_conns.erase(it);
		if (m_conns.empty() && m_bScheduled)
		{
			m_pReactor->CancelTimer(this);
			m_bScheduled = FALSE;
		}
	}
#endif

protected: // from IRtTimerHandler
	virtual void OnTimeout(const CRtTimeValue &aCurTime, LPVOID aArg)
	{
		RtRudpTimeUS time = GetRtRudpTimeUS();
#ifdef RUDP_CONNS_USE_ARRAY
		RtResult rv;
		unsigned i=0, sum = m_connArraySize; 
		if (m_nextConnIndex>=m_connArraySize)
			m_nextConnIndex = 0;

		while (i++<sum)
		{
			CRtRudpConn* pConn = m_connArray[m_nextConnIndex];
			if(pConn->m_status!=RUDP_STATE_CLOSED) 
			{
				rv = pConn->RunSendCycle(time);
// 				if (rv == RT_ERROR_PARTIAL_DATA)
// 					return;

#ifdef RUDP_FLOWCONTROL_OPTIMIZE
				// Check if need remove it from flow control and continue to handle next connection. 
				if ( !pConn->RemoveFromFlowControl() )
				{
					++m_nextConnIndex;
				}
#else
				++m_nextConnIndex;
#endif
			}
			else
				RemoveConn(pConn);

			if (m_nextConnIndex==m_connArraySize)
				m_nextConnIndex = 0;
		}
/*
		static RtRudpTimeUS ts = 0;
		if (time>ts+20000000)
		{
			printf("current connections: %d \n", m_connArraySize);
			ts = time;
		}
*/
#else
		RudpConnsType::iterator it=m_conns.begin();
		while (it!=m_conns.end())
		{
			RudpConnsType::iterator itTmp=it++;
			if (itTmp->second->m_status==RUDP_STATE_CLOSED)
				// Remove the conn object right here to avoid removing it in the call of 
				// CRtRudpConn::Disconnect() which would lead to crashing.
				RemoveConn(itTmp);
			else
				itTmp->second->RunSendCycle(time);
		}
#endif
	}

protected:
	CRtThread* m_pThreadNetwork;
	IRtReactor* m_pReactor;
#ifdef RUDP_CONNS_USE_ARRAY
 	RudpConnIndexesType m_connIndexes;
	CRtRudpConn** m_connArray;
	unsigned m_connArraySize;
	unsigned m_connArrayAllocSize;
	unsigned m_nextConnIndex;
#else
	RudpConnsType m_conns;
#endif
	BOOL m_bScheduled;

public:
	CRtRudpPacketPool* m_internalPacketPool;
};

// typedef CRtSingletonT<CRtRudpFlowControl> CRtRudpFlowControlSingleton;

RtResult CRtRudpFlowControlReleaseEvent::OnEventFire()
{
	m_instance->Release();
	return RT_OK;
}

#endif	//__RTCN_RUDPFLOWCONTROL_H__
